package TableAndSQL

import Source.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.table.api.Table
import org.apache.flink.table.api.scala._

object test01 {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val filePath = "src/main/resources/SensorReading"
    val inputStream = env.readTextFile(filePath)

    //转换成样例类
    val dataStream = inputStream
      .map(data => {
        val arr = data
          .split(",")
        SensorReading(arr(0), arr(1).toLong, arr(2).toDouble)
      })

    /**
     * Table API的使用
     */
    //创建表的执行环境
    val tableEnv = StreamTableEnvironment.create(env)

    //基于流创建表
    val dataTable = tableEnv.fromDataStream(dataStream)

    //调用table API转换
    val resultTable = dataTable
      .select("id, temperature")
      .filter("id == 'sensor_1'")

    resultTable.toAppendStream[(String, Double)].print("result")
    /**
     * 直接使用SQL实现
     */
    //创建表
    tableEnv.createTemporaryView("data", dataTable)
    val sql = "select id,temperature from data where id = 'sensor_1'"
    val resutlSQLTable = tableEnv.sqlQuery(sql)
    resutlSQLTable.toAppendStream[(String, Double)].print("SQLResult")
    env.execute()
  }
}
